package com.cwz.mq.ack.aspect;

import com.cwz.core.utils.string.StringUtils;
import com.cwz.mq.ack.config.RabbitConfig;
import com.cwz.mq.ack.constant.MsgLogConstants;
import com.cwz.mq.ack.entity.MsgLog;
import com.cwz.mq.ack.service.MsgLogService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


/**
 * @program: w-demo
 * @description: mq 消费者消息确认切面处理（用于手动消息确认，修改数据库的消息日志相关内容）
 * @author: Wen
 **/
@Aspect
@Component
@Slf4j
public class MqAckAspect {
	@Autowired
	private MsgLogService msgLogService;

	/**
	 * 定义 AOP 签名 (切入所有使用鉴权注解的方法)
	 */
	@Pointcut("@annotation(com.cwz.mq.ack.annotation.MqAck)")
	public void pointcut() {}

	/**
	 * 环绕切入
	 *
	 * @param joinPoint 切面对象
	 * @return 底层方法执行后的返回值
	 * @throws Throwable 底层方法抛出的异常
	 */
	@Around("pointcut()")
	public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
		// 睡 200ms，保证先触发交换机回调再处理队列逻辑
		Thread.sleep(200);
		Object[] args = joinPoint.getArgs();
		Message message = (Message) args[0];
		Channel channel = (Channel) args[1];

		String correlationId = RabbitConfig.getCorrelationId(message);
		// 消费幂等性, 防止消息被重复消费
		if (isConsumed(correlationId)) {
			return null;
		}
		long tag = message.getMessageProperties().getDeliveryTag();
		try {
			// 业务逻辑
			Object result = joinPoint.proceed();
			this.msgLogService.updateMsgStatus(correlationId, MsgLogConstants.CONSUMED_SUCCESS);
			// basicAck 方法是肯定的交付，一般在该消息处理完后执行，该消息才会在队列里面被删除，不然会处于 UnAcked 的状态存在队列中。
			// 其方法有两个参数：
			// 参数1（deliveryTag）：表示消息投递序号。
			// 参数2（multiple）: 是否批量确认消息，前提是在同一个 channel 里面，且是在该消息确认前没有被确认的消息才能批量确认。
			channel.basicAck(tag, false);
			return result;
		} catch (Exception e) {
			e.printStackTrace();
			log.error("MqAckAspect error: {}", e.getMessage());
			// 重新投递，重试次数 + 1
			this.msgLogService.addMsgTryCount(correlationId, null);
			// basicNack：表示失败确认，一般在消费消息业务异常时用到此方法，可以将消息重新投递入队列。
			// 其方法有三个参数：
			// 参数1（deliveryTag）：表示消息投递序号。
			// 参数2（multiple）：是否批量确认。
			// 参数3（requeue）：值为 true 消息将重新入队列。
			channel.basicNack(tag, false, true);
			return null;
		}
	}

	/**
	 * @Description: 消息是否已被消费/重试次数已达上限
	 * @Author: Wen
	 * @param correlationId:
	 * @return: boolean
	 **/
	private boolean isConsumed(String correlationId) {
		MsgLog msgLog = this.msgLogService.selectByMsgId(correlationId);
		return StringUtils.isNull(msgLog) || msgLog.getStatus().equals(MsgLogConstants.CONSUMED_SUCCESS)
			|| msgLog.getTryCount() >= MsgLogConstants.MAX_TRY_COUNT;
	}
}
